package com.raven.play.stream.test.scaladsl.tutorial

import akka.actor.{Actor, ActorSystem, Props}
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Flow, Sink, Source}

object $07OperatorFusion extends App {

  implicit val system = ActorSystem("OperatorFusion")
  implicit val materializer = ActorMaterializer()
  val simpleSource = Source(1 to 10)
  val simpleFlow = Flow[Int].map(_ + 1)
  val simpleFlow2 = Flow[Int].map(_ * 10)
  val simpleSink = Sink.foreach[Int](println)
  // this runs on the SAME ACTOR
  //  simpleSource.via(simpleFlow).via(simpleFlow2).to(simpleSink).run()

  // operator/component FUSION
  class SimpleActor extends Actor {
    override def receive: Receive = {
      case x: Int =>
        // flow operations
        val x2 = x + 1
        val y = x2 + 10
        // sink operation
        println(y)
    }

  }
  //  this runs on the SAME ACTOR
  //  simpleSource.via(simpleFlow).via(simpleFlow2).to(simpleSink).run(
  //  operator/component FUSION

  val simpleActor = system.actorOf(Props[SimpleActor])
//  (1 to 10).foreach(simpleActor ! _)

  // complex flows:
  val complexFlow = Flow[Int].map { x =>
    // simulating a long computation
    Thread.sleep(1000)
    x + 1
  }
  val complexFlow2 = Flow[Int].map { x =>
    // simulating a long computation
    Thread.sleep(1000)
    x * 10
  }
  //  simpleSource.via(complexFlow).via(complexFlow2).to(simpleSink).run()

  // async boundary
  // 原先没有 async 时, 所有操作都是在同一个 actor 上执行的
  // async 将操作步骤划分到不同的 actor 中, 那么原先 2 秒处理一个元素的, 变成看上去只需要 1 秒
  //  simpleSource.via(complexFlow).async // runs on one actor
  //    .via(complexFlow2).async // runs on another actor
  //    .to(simpleSink) // runs on a third actor
  //    .run()

  // ordering guarantees
  Source(1 to 3)
    .map(e => {println(s"Flow A: $e"); e}).async
    .map(e => {println(s"Flow B: $e"); e}).async
    .map(e => {println(s"Flow C: $e"); e}).async
    .runWith(Sink.ignore)

  system.terminate()
}


